package com.anmi.transfer.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
import com.anmi.transfer.table.base.SyncTable;
import com.anmi.transfer.util.SpringContextUtil;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.Map;

/** 这里和t_order 相关的数据同步操作 */
@Component
@Slf4j
public class MysqlToEsCanalClient extends AbstractCanalClient implements ApplicationRunner {

  public MysqlToEsCanalClient() {}

  public MysqlToEsCanalClient(String destination) {
    super.destination = destination;
  }

  @Value("${canal.server.ip}")
  private String canalServerIp;

  @Value("${canal.server.port}")
  private int canalServerPort;

  @Value("${canal.destination}")
  private String canalDestination;

  @Value("${canal.username}")
  private String canalUsername;

  @Value("${canal.password}")
  private String canalPassword;

  @Override
  // 当开启es增量同步监听时，启动的监听线程
  protected void process() {
    int batchSize = 5 * 1024;
    while (running) {
      // 当监听线程启动时
      try {
        MDC.put("destination", destination);
        connector.connect();
        connector.subscribe();
        while (running) {
          Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
          long batchId = message.getId();
          int size = message.getEntries().size();
          if (batchId == -1 || size == 0) {
            // try {
            // Thread.sleep(1000);
            // } catch (InterruptedException e) {
            // }
          } else {
            invokeAllMethods(message);
          }
          if (batchId != -1) {
            connector.ack(batchId); // 提交确认
            // connector.rollback(batchId); // 处理失败, 回滚数据
          }
        }
      } catch (Exception e) {
        logger.error("process error!", e);
        try {
          Thread.sleep(1000L);
        } catch (InterruptedException e1) {
        }
      } finally {
        connector.disconnect();
        MDC.remove("destination");
      }
    }
  }

  private void invokeAllMethods(Message message) {
    Map<String, SyncTable> beanMap = SpringContextUtil.getBeanMap(SyncTable.class);
    if (beanMap.isEmpty()) {
      // you should implement the interface[com.anmi.transfer.table.base.SyncTable]
      // throw new RuntimeException
      return;
    }
    beanMap.forEach((k, v) -> v.handleData(message.getEntries()));
  }

  @Override
  public void run(ApplicationArguments args) throws Exception {
    logger.info("canal监听开始启动...");
    // 启动增量同步的监听
    startSync(canalServerIp, canalServerPort, canalDestination, canalUsername, canalPassword);
    logger.info("canal监听启动完成...");
  }

  /**
   * canal-client 直连canal server的启动方法
   *
   * @param ip canal-server 的IP
   * @param port canal-server 的端口
   * @param destination canal-server的一个订阅地址，类似于订阅模型的订阅地址
   * @param canalUsername canal作为从节点的用户名
   * @param canalPassword canal作为从节点的密码
   */
  protected void startSync(
      String ip, int port, String destination, String canalUsername, String canalPassword) {

    CanalConnector connector =
        CanalConnectors.newSingleConnector(
            new InetSocketAddress(ip, port), destination, canalUsername, canalPassword);

    final MysqlToEsCanalClient clientTest = new MysqlToEsCanalClient(destination);
    clientTest.setConnector(connector);
    clientTest.start();
    Runtime.getRuntime()
        // 在jvm销毁前要执行的程序
        .addShutdownHook(
            new Thread(
                () -> {
                  try {
                    logger.info("## stop the canal client");
                    clientTest.stop();
                  } catch (Throwable e) {
                    logger.warn("##something goes wrong when stopping canal:", e);
                  } finally {
                    logger.info("## canal client is down.");
                  }
                }));
  }
}
